package org.qloudgen.borg.pages;

import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.wicket.ajax.AjaxRequestTarget;
import org.apache.wicket.ajax.markup.html.AjaxLink;
import org.apache.wicket.ajax.markup.html.form.AjaxButton;
import org.apache.wicket.ajax.markup.html.navigation.paging.AjaxPagingNavigator;
import org.apache.wicket.markup.html.WebMarkupContainer;
import org.apache.wicket.markup.html.basic.Label;
import org.apache.wicket.markup.html.form.Button;
import org.apache.wicket.markup.html.form.Form;
import org.apache.wicket.markup.html.form.FormComponent;
import org.apache.wicket.markup.html.form.Radio;
import org.apache.wicket.markup.html.form.RadioGroup;
import org.apache.wicket.markup.html.form.TextArea;
import org.apache.wicket.markup.html.list.ListItem;
import org.apache.wicket.markup.html.list.PageableListView;
import org.apache.wicket.model.CompoundPropertyModel;
import org.apache.wicket.model.Model;
import org.apache.wicket.request.mapper.parameter.PageParameters;
import org.qloudgen.borg.CommonValues;
import org.qloudgen.borg.beans.Cluster;
import org.qloudgen.borg.beans.DataSourceBean;
import org.qloudgen.borg.metadb.DefaultImpl;
import org.qloudgen.borg.pages.common.BasePage;
import org.qloudgen.borg.pages.model.DataSourceModel;
import org.qloudgen.borg.utils.CommonUtils;

import com.google.common.collect.Lists;

@SuppressWarnings({"unchecked","rawtypes"})
public class DataSourceFuncListPage extends BasePage{

	/**
	 * @Author youngeryoyo
	 * youngeryoyo@gmail.com
	 */

	static Log log = LogFactory.getLog(DataSourceFuncListPage.class);

	private static final long serialVersionUID = 1L;
	private transient List<DataSourceBean> dataSources = Lists.newArrayList();
	private transient String buttonName;
	private transient String appNo,dsNo;
	private transient List<Cluster> clusters = Lists.newArrayList();
	private transient List<String> files = Lists.newArrayList();
	
	public DataSourceFuncListPage(final PageParameters pageParameters){
		
		super();
		this.pp = pageParameters;
	
		// get dsNo
		dsNo = this.pp.get("dsNo").toString();
		
		// get from metadb
		DefaultImpl.getDataSourceFuncs(dsNo,dataSources);
		
		// border and detail
		final WebMarkupContainer border = new WebMarkupContainer("border");
		border.setOutputMarkupId(true);
		this.add(border);
		
		final TextArea<String> detail = new TextArea<String>("detail");
		detail.setOutputMarkupId(true);
		detail.setVisible(false);
		detail.setDefaultModel(new Model<String>(""));
		border.add(detail);
		
		// add form 
		Form<?> form = new Form("form",new CompoundPropertyModel<DataSourceModel>(new DataSourceModel())) {
			
			private static final long serialVersionUID = 1L;

			protected void onSubmit() {
				
				DataSourceBean ds = ((DataSourceModel)getDefaultModelObject()).getGroup();
				buttonName = ((FormComponent<String>)findSubmittingButton()).getInputName();
				
				if (ds == null ) {	info(CommonValues.INFO_NO_SELECT);  return; } 
				
				appNo = ds.getAppNo();
				
				if (buttonName.equals("upload")) {

					if (upLoad(ds) == -1) { error(CommonValues.ERROR_UPLOAD_FILE); return; }
					
					// update status
					DefaultImpl.updateDataSourceFuncStatus(ds.setStatus(CommonValues.STATUS_UPLOAD));
					return;
					
				} 
				
				if (buttonName.equals("delete")) {
					DefaultImpl.deleteDataSourceFunc(ds);
					setResponsePage(DataSourceFuncListPage.class,pp );
				}

			}
		};
		this.add(form);
		
		// add group
		final RadioGroup<?> group=new RadioGroup<Object>("group"); 
		group.setOutputMarkupId(true);
		group.setRenderBodyOnly(false);
		form.add(group);
		
		// add func list
		final PageableListView<DataSourceBean> listView =new PageableListView<DataSourceBean>( "rows", dataSources, CommonValues.SHOW_LIST_NUMBER) {

			private static final long serialVersionUID = 1L;

			@Override
			protected void populateItem(ListItem<DataSourceBean> item) {
				final DataSourceBean ds = (DataSourceBean) item.getModelObject();
				item.add(new Radio<DataSourceBean>("radio", new Model<DataSourceBean>(ds)));
				item.add( new Label( "id", ds.getDsNo()));
				item.add( new Label( "name", ds.getDsName()));
				item.add( new Label( "funcName", ds.getFuncName()));
				item.add( new Label( "appName", ds.getAppName()));
				
				item.add( new AjaxLink("transformfile"){

					private static final long serialVersionUID = 1L;

					@Override
					public void onClick(AjaxRequestTarget target) {
						detail.setDefaultModelObject(ds.getTransformFile());
						detail.setVisible(true);
						target.add(border);
						
					}});			

				item.add( new Label( "status", CommonValues.statusNames.get(ds.getStatus())));
			}
		};
		listView.setOutputMarkupId(true);
		group.add(listView);
		
		// add navigator
		final AjaxPagingNavigator nav = new AjaxPagingNavigator("navigator", listView);
		nav.setOutputMarkupId(true);
		group.add(nav);
		
		// add button 
		Button add = new Button("add"){
			private static final long serialVersionUID = 1L;
			public void onSubmit() {
				setResponsePage(DataSourceFuncCreatePage.class,pageParameters);
			}
		};
		add.setDefaultFormProcessing(false);
		form.add(add);
		
		form.add(new Button("delete"));
		
		form.add(new AjaxButton("upload"){
			private static final long serialVersionUID = 1L;
			protected void onAfterSubmit(AjaxRequestTarget target, Form<?> form) {
				
				if (clusters.size() == 0 ) return;
				StringBuffer clusterNames = new StringBuffer();
				for (Cluster cluster:clusters){
					clusterNames.append(" "+cluster.getClusterName());
				}
				
				detail.setDefaultModelObject(CommonValues.INFO_UPLOAD_COMPLETED + clusterNames.toString() );
				detail.setVisible(true);
				target.add(border);
				target.add(group);
				
			}});
		
	}

	/*
	 * upload files to cluster available to app no
	 */
	private int upLoad(DataSourceBean ds) {
		
		// write content to local file
		int ret = 0;
		String localTransformFile = CommonValues.getTempDir()+"/"+ CommonValues.getDSTransformFileName(ds.getDsNo(),ds.getFuncNo());
		ret = CommonUtils.writeFileToLocal(localTransformFile, ds.getTransformFile());	
		
		files.clear();
		files.add(localTransformFile);
		
		// upload to hadoop 
		
			// get app available hadoop 
			DefaultImpl.getAppCluster(appNo, clusters);
			
			ret = CommonUtils.scpFilesToClusters(clusters, files);
			
		return ret;
	}

}
